#include "thread_pool.h"
#include <assert.h>

using namespace lakes;

ThreadPool::ThreadPool(int min_thread, int max_thread, bool isStart) {
    init(min_thread, max_thread);
    if (isStart) {
        start();
    }
}

ThreadPool::~ThreadPool() {
    finish();
}

void ThreadPool::start() {
    if (!m_isRunning) {
        m_isRunning = true;
        addManagerThread();
        for (int i = 0; i < this->getMinThreadNumber(); i++) {
            addNewWorkThread();
            std::this_thread::sleep_for(std::chrono::microseconds(10));
        }
    }
}

void ThreadPool::finish() {
    if (m_isRunning) {
        m_isRunning = false;
        m_taskQueue.signalAllConsumer();

        m_threadManager->join();
        delete m_threadManager;

        for (auto& val : m_threadWorkMap) {
            val.second->join();
            delete val.second;
        }

        if (m_threadHook) {
            m_threadHook->join();
            delete m_threadHook;
        }
    }
}

void ThreadPool::init(int min_thread, int max_thread) {
    if (min_thread > max_thread || min_thread <= 0) {
        assert(1);
    }
    m_minThreadNum = min_thread;
    m_maxThreadNum = max_thread;
    m_liveThreadNum = 0;
    m_isRunning = false;
    m_busyThreadNum = 0;
    m_willKillThreadNum = 0;
}

void lakes::ThreadPool::addManagerThread() {
    if (m_threadManager) {
        return;
    }
  
    m_threadManager = new std::thread([=]() {
            while (this->isRun()) {
                std::chrono::milliseconds dura(MANAGER_INTERVAL_MS);
                std::this_thread::sleep_for(dura);
                if (this->getLiveThreadNumber() > this->getMinThreadNumber() && this->getBusyThreadNumber() * 2 < this->getLiveThreadNumber()) {   
                    this->m_willKillThreadNum = ONCE_KILL_THREADA;
                    for (int i = 0; i < ONCE_KILL_THREADA; ++i) {
                        m_taskQueue.signalOneConsumer();
                    }
                }
                
                if (this->getLiveThreadNumber() <  this->getMaxThreadNumber() && this->getLiveThreadNumber() < this->getTaskQueueNumber()) {                   
                    for (int i = 0; i < ONCE_CREATE_THREADA &&  this->getLiveThreadNumber() <  this->getMaxThreadNumber(); ++i) {                        
                        this->addNewWorkThread();        
                    }
                }     
            }
    });
}

void lakes::ThreadPool::addNewWorkThread() {
    if (!isRun()) {
        return;
    } 
    std::thread* tmp = new std::thread([=]() {
        THREAD_STATU statu;
        while (1) {
            statu = WAITTING;
            while (this->getTaskQueueNumber() == 0 && this->isRun()) {  
                m_taskQueue.waitProducerSignal();
                if (this->m_willKillThreadNum > 0 && this->m_liveThreadNum > this->getMinThreadNumber()) {
                    this->m_willKillThreadNum--; 
                    statu = DEAD;
                    break;                     
                } 
            }  

            if (!this->isRun() || statu == DEAD) { 
                this->killSelfThread(statu == DEAD);
                break;
            }

            auto task = this->m_taskQueue.GetTask();
            this->m_busyThreadNum++;
            if (task) {
                task();
            }
            this->m_busyThreadNum--;    
        }
    });
    m_PoolMutex.lock();
    m_threadWorkMap.emplace(tmp->get_id(), tmp);  
    m_liveThreadNum++;
    m_PoolMutex.unlock();
}

void ThreadPool::killSelfThread(bool isCover) {
    std::lock_guard<std::mutex> lg(m_PoolMutex);
    m_liveThreadNum--;
    if (isCover) {
        m_threadWorkMap[std::this_thread::get_id()]->detach();
        if (m_threadWorkMap.count(std::this_thread::get_id())) {
            delete m_threadWorkMap[std::this_thread::get_id()];
            m_threadWorkMap.erase(std::this_thread::get_id());
        }
    }
}

int ThreadPool::getMinThreadNumber() {
    return m_minThreadNum;
}

int ThreadPool::getMaxThreadNumber() {
    return m_maxThreadNum;
}

int ThreadPool::getTaskQueueNumber() {
    return m_taskQueue.getTaskQueueNum();
}

int ThreadPool::getBusyThreadNumber() {
    return m_busyThreadNum;
}

int ThreadPool::getLiveThreadNumber() {
    return m_liveThreadNum;
}

void lakes::ThreadPool::setHook(int ms, std::function<void(ThreadPool *)> hookCallback) {
    m_threadHook = new std::thread([=]() {
        while (isRun()) {
          std::this_thread::sleep_for(std::chrono::milliseconds(ms));
          if (hookCallback) {
            hookCallback(this);
          }
        }
    });
}

bool ThreadPool::isRun() {
    return m_isRunning;
}

